package com.example.hbase.service;

import com.alibaba.lindorm.client.core.utils.Bytes;
import com.example.hbase.HBaseConstant;
import com.example.hbase.annotation.ColumnInfo;
import com.example.hbase.annotation.TypeInfo;
import com.example.hbase.handler.DefaultTypeHandlers;
import com.example.hbase.handler.TypeHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;

import java.io.IOException;
import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.util.*;

@Slf4j
public class HBaseService {
    /**
     * hbase 连接为重资源对象，自带连接池，线程安全可以复用，其他资源都必须随用随关
     */
    public static Connection connection = null;

    static {
        org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", HBaseConstant.lindormUrl);
        conf.set("hbase.client.username", HBaseConstant.username);
        conf.set("hbase.client.password", HBaseConstant.password);
        try {
            connection = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            log.error("获取HBase连接失败!");
        }
    }

    /**
     * hbase库存在的表
     * @return
     */
    public static List<String> listTable() {
        List<String> list = new ArrayList<>();
        try(Admin admin = connection.getAdmin()){
            for(TableName tn : admin.listTableNames()){
                list.add(tn.getNameAsString());
            }
        }catch (IOException e){
            log.error("HBaseService.listTable", e);
        }
        return list;
    }

    /**
     * 建表
     * @param tableName    表名
     * @param familyNames  列簇
     */
    public static void createTable(String tableName, String... familyNames){
        try(Admin admin = connection.getAdmin()){
            TableName table = TableName.valueOf(tableName);
            // 建表
            TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(table);
            List<ColumnFamilyDescriptor> familyList = new ArrayList<>();
            for (String familyName : familyNames) {
                ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor cf = (ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor)
                        ColumnFamilyDescriptorBuilder.of(Bytes.toBytes(familyName));
                cf.setMaxVersions(3);
                familyList.add(cf);
            }
            builder.setColumnFamilies(familyList);
            TableDescriptor descriptor = builder.build();
            if(!admin.tableExists(table)){
                admin.createTable(descriptor);
            }
        }catch (IOException e){
            log.error("HBaseService.createTable", e);
        }

    }

    /**
     * 无效表
     * @param tableName
     */
    public static void disableTable(String tableName) {
        try(Admin admin = connection.getAdmin()){
            TableName table = TableName.valueOf(tableName);
            if(admin.tableExists(table)){
                admin.disableTable(table);
            }
        }catch (IOException e){
            log.error("HBaseService.disableTable", e);
        }
    }

    /**
     * 启用表
     * @param tableName
     */
    public static void enableTable(String tableName){
        try(Admin admin = connection.getAdmin()){
            TableName table = TableName.valueOf(tableName);
            if(admin.tableExists(table)){
                admin.enableTable(table);
            }
        }catch (IOException e){
            log.error("HBaseService.enableTable", e);
        }
    }

    /**
     * 删除表
     * @param tableName
     */
    public static void dropTable(String tableName){
        try(Admin admin = connection.getAdmin()){
            TableName table = TableName.valueOf(tableName);
            if(admin.tableExists(table)){
                // 先无效
                admin.disableTable(table);
                // 删除表
                admin.deleteTable(table);
            }
        }catch (IOException e){
            log.error("HBaseService.dropTable", e);
        }
    }

    /**
     * 清空表数据
     * @param tableName
     */
    public static void truncateTable(String tableName) {
        try(Admin admin = connection.getAdmin()){
            TableName table = TableName.valueOf(tableName);
            if(admin.tableExists(table)){
                // truncate 表
                admin.truncateTable(table, true);
            }
        }catch (IOException e){
            log.error("HBaseService.truncateTable", e);
        }
    }

    /**
     * 插入数据
     * @param tableName
     * @param rowKey
     * @param family
     * @param qualifier
     * @param value
     * @throws RuntimeException
     */
    public static void insert(String tableName, String rowKey, String family, String qualifier, Object value) throws RuntimeException{
        byte[] bytes;
        if(value instanceof Short){
            bytes = Bytes.toBytes((Short) value);
        }else if(value instanceof Integer){
            bytes = Bytes.toBytes((Integer) value);
        }else if(value instanceof Long){
            bytes = Bytes.toBytes((Long) value);
        }else if(value instanceof Float){
            bytes = Bytes.toBytes((Float) value);
        }else if(value instanceof Double){
            bytes = Bytes.toBytes((Double) value);
        }else if(value instanceof String){
            bytes = Bytes.toBytes((String) value);
        }else if(value instanceof Boolean){
            bytes = Bytes.toBytes((Boolean) value);
        }else if(value instanceof BigDecimal){
            bytes = Bytes.toBytes((BigDecimal) value);
        }else{
            throw new RuntimeException("暂不支持插入hbase的数据类型");
        }
        Put put = new Put(Bytes.toBytes(rowKey));
        put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), bytes);
        insert(tableName, put);
    }

    /**
     * 插入数据
     * @param tableName
     * @param put
     */
    public static void insert(String tableName, Put put){
        TableName tableNameObj = TableName.valueOf(tableName);
        try (Table table = connection.getTable(tableNameObj)) {
            // 插入数据
            table.put(put);
        }catch (IOException e){
            log.error("HBaseService.insert", e);
        }
    }

    /**
     * get方式获取数据
     * @param tableName
     * @param get
     * @return
     */
    public static Result getData(String tableName, Get get){
        Result result = new Result();
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            // 单行读取
            result = table.get(get);
        }catch (IOException e){
            log.error("HBaseService.getData", e);
        }
        return result;
    }
    
    /**
     * 获取一列获取一行数据
     * @param tableName
     * @param rowKey
     * @param famliyName
     * @param column
     * @return
     * @throws IOException
     */
    public static String getData(String tableName, String rowKey, String famliyName, String column) throws IOException {
        byte[] row = Bytes.toBytes(rowKey);
        Get get = new Get(row);
        byte[] resultValue = null;
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            // 单行读取
            Result res = table.get(get);
            resultValue = res.getValue(famliyName.getBytes(), column.getBytes());
        }catch (IOException e){
            log.error("HBaseService.getData", e);
        }
        if(null == resultValue){
            return null;
        }
        return new String(resultValue);
    }

    /**
     * 获取对象值
     * @param tableName
     * @param rowKey
     * @param type
     * @param <T>
     * @return
     * @throws Exception
     */
    public static <T> T getObject(String tableName, String rowKey, Class<T> type) throws Exception {
        TypeInfo typeInfo = TypeInfo.parse(type);

        try(Table table = connection.getTable(TableName.valueOf(tableName))){
            T obj = type.newInstance();
            
            Get get = new Get(Bytes.toBytes(rowKey));
            Result r = table.get(get);
            for (Cell cell : r.listCells()){
                ColumnInfo columnInfo = typeInfo.findColumnInfo(Bytes.toString(CellUtil.cloneFamily(cell)), Bytes.toString(CellUtil.cloneQualifier(cell)));
                if(columnInfo==null){
                    continue;
                }
                TypeHandler typeHandler = DefaultTypeHandlers.findDefaultHandler(columnInfo.getField().getType());
                Object value = typeHandler.toObject(columnInfo.getField().getType(), CellUtil.cloneValue(cell));

                if (value != null) {
                    columnInfo.getField().set(obj, value);
                }
            }

            byte[] rowKeys = r.getRow();
            ColumnInfo rowKeyColumn = typeInfo.getRowKey();
            if(rowKeyColumn!=null){
                rowKeyColumn.getField().set(obj, new String(rowKeys));
            }
            return obj;
        }
    }

    public static <T> List<T> listObject(String tableName, Scan scan, Class<T> type) throws Exception {
        List<T> list = new ArrayList<>();
        
        TypeInfo typeInfo = TypeInfo.parse(type);

        try(Table table = connection.getTable(TableName.valueOf(tableName))){
            ResultScanner scanner = table.getScanner(scan);
            for (Result r : scanner) {
                T obj = type.newInstance();
                for (Cell cell : r.listCells()){
                    ColumnInfo columnInfo = typeInfo.findColumnInfo(Bytes.toString(CellUtil.cloneFamily(cell)),
                            Bytes.toString(CellUtil.cloneQualifier(cell)));
                    if(columnInfo==null){
                        continue;
                    }
                    TypeHandler typeHandler = DefaultTypeHandlers.findDefaultHandler(columnInfo.getField().getType());
                    Object value = typeHandler.toObject(columnInfo.getField().getType(), CellUtil.cloneValue(cell));

                    if (value != null) {
                        columnInfo.getField().set(obj, value);
                    }
                }

                byte[] rowKeys = r.getRow();
                ColumnInfo rowKeyColumn = typeInfo.getRowKey();
                if(rowKeyColumn!=null){
                    rowKeyColumn.getField().set(obj, new String(rowKeys));
                }
                list.add(obj);
            }
        }
        return list;
    }
    
    /**
     * 唯一数据标识
     * rowkey+column+timestamp+type+value
     */
    public static List<Cell> listDataCell(String tableName, String rowKey){
        Get get = new Get(Bytes.toBytes(rowKey));
        List<Cell> list = listDataCell(tableName, get);
        return list;
    }
    
    /**
     * 唯一数据标识
     * rowkey+column+timestamp+type+value
     */
    public static List<String> listData(String tableName, String rowKey){
        List<String> valueList = new ArrayList<>();
        List<Cell> list = listDataCell(tableName, rowKey);
        for (Cell cell : list) {
            long timestamp = cell.getTimestamp();
            Cell.Type type = cell.getType();
            String value = Bytes.toString(CellUtil.cloneValue(cell));
            log.debug("rowKey:{},列族:{},列:{},时间戳:{}, 方法类型:{}, value:{}", 
                    Bytes.toString(CellUtil.cloneRow(cell)), 
                    Bytes.toString(CellUtil.cloneFamily(cell)),
                    Bytes.toString(CellUtil.cloneQualifier(cell)), 
                    timestamp, type, value);
            valueList.add(value);
        }
        return valueList;
    }

    /**
     * get方式获取数据
     * @param tableName
     * @param get
     * @return
     */
    public static List<Cell> listDataCell(String tableName, Get get){
        List<Cell> cells = new ArrayList<>();
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            // 单行读取
            Result res = table.get(get);
            cells = res.listCells();
        }catch (IOException e){
            log.error("HBaseService.getData", e);
        }
        return cells;
    }

    /**
     * 获取int类型返回
     * @param tableName
     * @param rowKey
     * @param family
     * @param qualifier
     * @return
     */
    public static Integer getIntegerValue(String tableName, String rowKey, String family, String qualifier) {
        Get get = new Get(rowKey.getBytes());
        Result result = getData(tableName, get);
        byte[] bytes = result.getValue(family.getBytes(), qualifier.getBytes());
        return Bytes.toInt(bytes);
    }
    
    /**
     * 获取long类型返回
     * @param tableName
     * @param rowKey
     * @param family
     * @param qualifier
     * @return
     */
    public static Long getLongValue(String tableName, String rowKey, String family, String qualifier) {
        Get get = new Get(rowKey.getBytes());
        Result result = getData(tableName, get);
        byte[] bytes = result.getValue(family.getBytes(), qualifier.getBytes());
        return Bytes.toLong(bytes);
    }

    /**
     * 查询rowkey下某一个字段值
     * @param tableName
     * @param rowKey
     * @param family
     * @param qualifier
     * @return
     */
    public static String getValue(String tableName, String rowKey, String family, String qualifier) {
        Table table = null;
        try {
            table = connection.getTable(TableName.valueOf(tableName));
            Get get = new Get(rowKey.getBytes());
            //返回指定列族、列名，避免rowKey下所有数据
            get.addColumn(family.getBytes(), qualifier.getBytes());
            Result rs = table.get(get);
            Cell cell = rs.getColumnLatestCell(family.getBytes(), qualifier.getBytes());
            String value = null;
            if (cell!=null) {
                value = Bytes.toString(CellUtil.cloneValue(cell));
            }
            return value;
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (table!=null){
                try {
                    table.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        return null;
    }

    /**
     * 获取一行的所有数据 并且排序
     * @param tableName 表名
     * @param rowKey 列名
     * @throws IOException
     */
    public static List<Map.Entry> getRow(String tableName, String rowKey) throws IOException {
        List<Map.Entry> ans = new ArrayList<>();
        try(Table table = connection.getTable(TableName.valueOf(tableName))){
            byte[] row = Bytes.toBytes(rowKey);
            Get get = new Get(row);
            Result r = table.get(get);
            HashMap<String, Double> rst = new HashMap<>();

            for (Cell cell : r.listCells()){
                String key = org.apache.hadoop.hbase.util.Bytes.toString(cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength());
                String value = org.apache.hadoop.hbase.util.Bytes.toString(cell.getValueArray(),cell.getValueOffset(),cell.getValueLength());
                rst.put(key, new Double(value));
            }
            ans.addAll(rst.entrySet());

            Collections.sort(ans, (m1, m2) -> new Double((Double)m1.getValue()-(Double) m2.getValue()).intValue());
        }
        return ans;
    }

    /**
     * 向对应列添加数据
     * @param tablename 表名
     * @param rowkey 行号
     * @param familyName 列族名
     * @param column 列名
     * @param data 数据
     * @throws Exception
     */
    public static void putData(String tablename, String rowkey, String familyName,String column,String data) throws Exception {
        try(Table table = connection.getTable(TableName.valueOf(tablename));){
            Put put = new Put(rowkey.getBytes());
            put.addColumn(familyName.getBytes(),column.getBytes(),data.getBytes());
            table.put(put);
        }
    }

    /**
     * 添加数据
     * @param tableName
     * @param rowkey
     * @param familyName
     * @param column
     * @param timestamp
     * @param data
     * @throws Exception
     */
    public static void putData(String tableName, String rowkey, String familyName,String column, Long timestamp, String data) throws Exception {
        try(Table table = connection.getTable(TableName.valueOf(tableName))){
            Put put = new Put(rowkey.getBytes());
            put.addColumn(familyName.getBytes(), column.getBytes(), timestamp, data.getBytes());
            table.put(put);
        }
    }

    /**
     * 检查db值是否存在，存在
     * @param tableName
     * @param rowkey
     * @param familyName
     * @param column
     * @param newDate
     * @param oldData
     * @throws Exception
     */
    public static boolean putAndCheck(String tableName, String rowkey, String familyName,String column, String newDate, String oldData) throws Exception {
        try(Table table = connection.getTable(TableName.valueOf(tableName))){
            Put put = new Put(rowkey.getBytes());
            put.addColumn(familyName.getBytes(), column.getBytes(), newDate.getBytes());
            
            Table.CheckAndMutateBuilder checkAndMutateBuilder = table.checkAndMutate(rowkey.getBytes(), familyName.getBytes());
            boolean result = checkAndMutateBuilder.qualifier(column.getBytes(StandardCharsets.UTF_8)).ifEquals(oldData.getBytes()).thenPut(put);
            return result;
        }
    }

    /**
     * 添加对象数据
     * @param tableName
     * @param rowkey
     * @param familyName
     * @param object
     * @throws Exception
     */
    public static void putObject(String tableName, String rowkey, String familyName, Object object) throws Exception {
        TypeInfo typeInfo = TypeInfo.parse(object.getClass());
        List<ColumnInfo> list = typeInfo.getColumnInfos();

        try(Table table = connection.getTable(TableName.valueOf(tableName))){
            Put put = new Put(Bytes.toBytes(rowkey));
            for (ColumnInfo columnInfo : list) {
                Class<?> fieldType = columnInfo.getField().getType();
                String fieldName = columnInfo.getField().getName();
                TypeHandler typeHandler = DefaultTypeHandlers.findDefaultHandler(fieldType);
                byte[] bytes = typeHandler.toBytes(fieldType, columnInfo.getField().get(object));
                if(bytes==null || bytes.length==0){
                    continue;
                }
                put.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(fieldName), bytes);
            }
            table.put(put);
        }
    }

    /**
     * 将该单元格加1
     * @param tablename 表名
     * @param rowkey 行号
     * @param famliyname 列族名
     * @param column 列名
     * @throws Exception
     */
    public static void incStringColumn(String tablename, String rowkey, String famliyname,String column) throws Exception {
        String val = getData(tablename, rowkey, famliyname, column);
        int res = 1;
        if (val != null) {
            res = Integer.valueOf(val) + 1;
        }
        putData(tablename, rowkey, famliyname, column, String.valueOf(res));
    }

    /**
     * 追加数据
     * @param tablename
     * @param rowkey
     * @param familyName
     * @param column
     * @param data
     * @throws Exception
     */
    public static void appendData(String tablename, String rowkey, String familyName,String column,String data) throws Exception {
        try(Table table = connection.getTable(TableName.valueOf(tablename));){
            Append append = new Append(rowkey.getBytes());
            append.addColumn(familyName.getBytes(),column.getBytes(),data.getBytes());
            table.append(append);
        }
    }

    /**
     * 数据自增
     * @param tablename
     * @param rowkey
     * @param familyName
     * @param column
     * @param data
     * @throws Exception
     */
    public static void increaseData(String tablename, String rowkey, String familyName,String column,long data) throws Exception {
        try(Table table = connection.getTable(TableName.valueOf(tablename));){
            Increment increment = new Increment(rowkey.getBytes());
            increment.addColumn(familyName.getBytes(),column.getBytes(), data);
            table.increment(increment);
        }
    }
    
    /**
     * 取出表中所有的key
     * @param tableName
     * @return
     */
    public static List<String> getAllKey(String tableName) throws IOException {
        List<String> keys = new ArrayList<>();
        Scan scan = new Scan();
        try(Table table = connection.getTable(TableName.valueOf(tableName))){
            ResultScanner scanner = table.getScanner(scan);
            for (Result r : scanner) {
                keys.add(new String(r.getRow()));
            }
        }
        return keys;
    }

    /**
     * 删除重载的底层方法
     * @param tableName
     * @param delete
     */
    public static void delete(String tableName, Delete delete){
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            table.delete(delete);
        }catch (Exception e){
            log.error("HBaseService.delete", e);
        }
    }

    /**
     * 删除数据
     * @param tableName
     * @param rowKey
     */
    public static void deleteRow(String tableName, String rowKey){
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        delete(tableName, delete);
    }

    /**
     * 删除数据
     * @param tableName
     * @param rowKey
     */
    public static void deleteColumn(String tableName, String rowKey, String familyName, String column){
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        delete.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(column));
        delete(tableName, delete);
    }

    /**
     * 扫描重载的底层方法
     * @param tableName
     * @param scan
     * @return
     */
    public static List<Result> scanData(String tableName, Scan scan){
        List<Result> list = new ArrayList<>();
        try (Table table = connection.getTable(TableName.valueOf(tableName)); 
             ResultScanner scanner = table.getScanner(scan)) {
            // scan 范围数据
            for (Result result : scanner) {
                for(Cell cell : result.listCells()){
                    log.debug("rowKey:{},列族:{},时间戳:{},列:{},value:{}", 
                            Bytes.toString(CellUtil.cloneRow(cell)), 
                            Bytes.toString(CellUtil.cloneFamily(cell)),
                            cell.getTimestamp(),
                            Bytes.toString(CellUtil.cloneQualifier(cell)), 
                            Bytes.toString(CellUtil.cloneValue(cell)));
                }
                list.add(result);
            }
        }catch (Exception e){
            log.error("HBaseService.scanData", e);
        }
        return list;
    }
}
